package net.lmc.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


public class LogSplitProcessFunction extends ProcessFunction<String,String> {

    /*
    侧输出流输出时标记OutputTag实例
     */
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public LogSplitProcessFunction(OutputTag<String> errorTag,OutputTag<String> startTag,OutputTag<String> displayTag,OutputTag<String> actionTag){
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String s, Context ctx, Collector<String> out) throws Exception {
        //diybu，解析JSON数据
        JSONObject jsonObject = JSON.parseObject(s);

        //第二步，错误日志
        if(null != jsonObject.get("err")){
            //侧边流输出
            ctx.output(errorTag,s);
            jsonObject.remove("err");
        }

        //第三步，启动日志
        if(null != jsonObject.get("start")){
            //侧边流输出
            ctx.output(startTag,jsonObject.toJSONString());
            //第四步，页面日志
        }else{
            //获取曝光和动作数据
            JSONArray displaysJsonArray = jsonObject.getJSONArray("displays");
            JSONArray actionsJsonArray = jsonObject.getJSONArray("actions");

            //删除曝光和动作
            jsonObject.remove("displays");
            jsonObject.remove("actions");

            //输出页面日志
            out.collect(jsonObject.toJSONString());

            //第五步，曝光日志
            if(null != displaysJsonArray && !displaysJsonArray.isEmpty()){
                for (int i = 0; i < displaysJsonArray.size(); i ++){
                    JSONObject displayJsonObject = displaysJsonArray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("display",displayJsonObject);
                    //输出
                    ctx.output(displayTag,jsonObject.toJSONString());
                }
            }

            jsonObject.remove("display");//将上述添加曝光数据字段值删除

            //第六步，动作日志
            if(null != actionsJsonArray && !actionsJsonArray.isEmpty()){
                for(int  i = 0; i < actionsJsonArray.size(); i ++){
                    JSONObject actionJsonObject = actionsJsonArray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("action",actionJsonObject);
                    //输出
                    ctx.output(actionTag,jsonObject.toJSONString());
                }
            }
        }
    }
}
